package com.atguigu.flink.chapter09_cep;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/10/29
 *
 *
 *
 */
public class Demo8_ProcessTimeOut
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((element, recordTimestamp) -> element.getTs());

        //①有一个流，在流上设置水印(基于事件时间)，ProcessingTime不支持
        SingleOutputStreamOperator<WaterSensor> ds = env
            .readTextFile("data/cep2.txt").setParallelism(1)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy);

        /*
                匹配规则的第一条数据，属于超时数据。

                CEP中数据分为三类:
                        1.符合匹配规则的数据
                            {匹配s1=[WaterSensor(id=s1, ts=5000, vc=20)], 匹配s2=[WaterSensor(id=s2, ts=6999, vc=20)]}
                        2.不符合规则的数据
                                取不到
                        3.符合规则但是超时的数据
                                可以获取

         */
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                                          .<WaterSensor>begin("匹配s1")   //规则名为  匹配s1
                                          .where(new SimpleCondition<WaterSensor>()
                                          {
                                              //定义过滤规则
                                              @Override
                                              public boolean filter(WaterSensor value) throws Exception {
                                                  return "s1".equals(value.getId());
                                              }
                                          })
                                          .next("匹配s2")
                                            .where(new SimpleCondition<WaterSensor>()
                                            {
                                                @Override
                                                public boolean filter(WaterSensor value) throws Exception {
                                                    return "s2".equals(value.getId());
                                                }
                                            })
                                            .within(Time.seconds(2))  // < 2
                                            ;

        //③调用CEP的库，把pattern作用到流上，得到 模式流(存放匹配到的数据)
        PatternStream<WaterSensor> ds2 = CEP.pattern(ds, pattern);

        OutputTag<String> outputTag = new OutputTag<>("timeout", Types.STRING);
        //④从模式流中获取匹配的数据

        SingleOutputStreamOperator<String> ds3 = ds2.process(new MyProcess());

        ds3.print("匹配").setParallelism(1);
        ds3.getSideOutput(outputTag).print("超时").setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    public static class MyProcess extends PatternProcessFunction<WaterSensor, String> implements TimedOutPartialMatchHandler<WaterSensor>{

        //处理匹配到
        @Override
        public void processMatch(Map<String, List<WaterSensor>> match, Context ctx, Collector<String> out) throws Exception {

            out.collect(match.toString());
        }

        //处理超时
        @Override
        public void processTimedOutMatch(Map<String, List<WaterSensor>> match, Context ctx) throws Exception {

            //直接使用侧流输出
                ctx.output(new OutputTag<>("timeout", Types.STRING),match.toString());
        }
    }
}
